package org.example;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Flink 流处理与批处理基本实现示例
 * 本示例展示如何使用 Flink 的 DataStream API 进行流处理，以及使用 DataSet API 进行批处理。
 * 流处理：从文本流中统计单词出现次数，实时输出结果。
 * 批处理：从文本文件中读取数据，统计单词出现次数，最后输出结果。
 * 适合初学者学习 Flink 的基本使用。
 */
public class StreamBatchExample {

    public static void main(String[] args) throws Exception {
        // 流处理示例
        streamProcessingExample();
        // 批处理示例
        batchProcessingExample();
    }

    /**
     * 流处理示例：从文本流中统计单词出现次数
     * 使用 DataStream API 实现实时数据处理。
     */
    private static void streamProcessingExample() throws Exception {
        // 创建流处理执行环境
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为 1，方便观察输出
        streamEnv.setParallelism(1);

        // 模拟输入数据流（实际应用中可能来自 Kafka、Socket 等）
        DataStream<String> textStream = streamEnv.fromElements(
            "Hello Flink",
            "Hello World",
            "Hello Flink"
        );

        // 使用 FlatMap 将每行文本拆分为单词，并转换为 (word, 1) 的元组
        DataStream<Tuple2<String, Integer>> wordCountStream = textStream
            .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public void flatMap(String line, Collector<Tuple2<String, Integer>> out) {
                    for (String word : line.split(" ")) {
                        out.collect(new Tuple2<>(word, 1));
                    }
                }
            })
            // 按单词分组，并累加计数
            .keyBy(0)
            .sum(1);

        // 输出结果
        wordCountStream.print("流处理结果");

        // 执行流处理作业
        streamEnv.execute("Stream Word Count Example");
    }

    /**
     * 批处理示例：从文本文件中读取数据，统计单词出现次数
     * 使用 DataSet API 实现离线数据处理。
     */
    private static void batchProcessingExample() throws Exception {
        // 创建批处理执行环境
        ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为 1，方便观察输出
        batchEnv.setParallelism(1);

        // 模拟输入数据（实际应用中可能从文件、数据库等读取）
        DataSet<String> textDataSet = batchEnv.fromElements(
            "Hello Flink",
            "Hello World",
            "Hello Flink"
        );

        // 使用 FlatMap 将每行文本拆分为单词，并转换为 (word, 1) 的元组
        DataSet<Tuple2<String, Integer>> wordCountDataSet = textDataSet
            .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public void flatMap(String line, Collector<Tuple2<String, Integer>> out) {
                    for (String word : line.split(" ")) {
                        out.collect(new Tuple2<>(word, 1));
                    }
                }
            })
            // 按单词分组，并累加计数
            .groupBy(0)
            .sum(1);

        // 输出结果
        wordCountDataSet.print("批处理结果");

        // 执行批处理作业
        batchEnv.execute("Batch Word Count Example");
    }
} 